package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime.bean.CourseUserReviewBean;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.MyClickhouseUtil;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @Author zhangsan
 * @Date 2022/10/19 14:45
 * @Description //TODO dws 课程用户 评价
 */
public class DwsCourseUserReview {
    public static void main(String[] args) throws Exception{
        //TODO 1.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //TODO 2.检查点设置

        //TODO 3.从kafka 的 课程用户评价主题中读取数据 dwd_course_user_review_info
        String topic = "dwd_course_user_review_info";
        String groupId = "dws_course_user_review";
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);
        //TODO 4.对读取的数据进行类型转换 jsonStr -> jsonObj  再转化为实体类  CourseUserReviewBean
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);
        SingleOutputStreamOperator<CourseUserReviewBean> reviewBean = jsonObjDS.map(
//                {"id":"18544","user_id":"19","course_id":"273","review_stars":"5","ts":"2022-10-18 20:15:26"}
//                {"id":"18551","user_id":"35","course_id":"374","review_stars":"5","ts":"2022-10-18 23:17:11"}
//                {"id":"18555","user_id":"15","course_id":"443","review_stars":"5","ts":"2022-10-18 19:27:37"}

                new MapFunction<JSONObject, CourseUserReviewBean>() {
                    @Override
                    public CourseUserReviewBean map(JSONObject jsonObj) throws Exception {
                        String courseId = jsonObj.getString("course_id");
                        String userId = jsonObj.getString("user_id");
                        String reviewStars = jsonObj.getString("review_stars");
                        Long ts = jsonObj.getLong("ts");

                        CourseUserReviewBean reviewBean = CourseUserReviewBean.builder()
                                .courseId(courseId)
                                .userId(userId)
                                .userCnt(1L)
                                .reviewStars(Long.valueOf(reviewStars))
                                .fiveStarsCnt("5".equals(reviewStars) ? 1L : 0L)
                                .ts(ts)
                                .build();

                        return reviewBean;

                    }
                }
        );
        //TODO 5.指定watermark以及提取事件时间字段
        SingleOutputStreamOperator<CourseUserReviewBean> withWatermarkDS = reviewBean.assignTimestampsAndWatermarks(
                WatermarkStrategy.<CourseUserReviewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<CourseUserReviewBean>() {
                                    @Override
                                    public long extractTimestamp(CourseUserReviewBean reviewBean, long l) {
                                        return reviewBean.getTs();
                                    }
                                }
                        )
        );
        //TODO 6.按照课程id进行分组
        KeyedStream<CourseUserReviewBean, String> courseIdKeyedDS = withWatermarkDS.keyBy(CourseUserReviewBean::getCourseId);

        //TODO 7.开窗
        WindowedStream<CourseUserReviewBean, String, TimeWindow> windowDS = courseIdKeyedDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));

        //TODO 8.聚合
        SingleOutputStreamOperator<CourseUserReviewBean> reduceDS = windowDS.reduce(
                new ReduceFunction<CourseUserReviewBean>() {
                    @Override
                    public CourseUserReviewBean reduce(CourseUserReviewBean value1, CourseUserReviewBean value2) throws Exception {
                        value1.setUserCnt(value1.getUserCnt() + value2.getUserCnt());
                        value1.setReviewStars(value1.getReviewStars() + value2.getReviewStars());
                        value1.setFiveStarsCnt(value1.getFiveStarsCnt() + value2.getFiveStarsCnt());
                        return value1;
                    }
                },
                new WindowFunction<CourseUserReviewBean, CourseUserReviewBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<CourseUserReviewBean> input, Collector<CourseUserReviewBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(window.getStart());
                        String edt = DateFormatUtil.toYmdHms(window.getEnd());

                        for (CourseUserReviewBean reviewBean : input) {
                            reviewBean.setStt(stt);
                            reviewBean.setEdt(edt);
                            reviewBean.setTs(System.currentTimeMillis());
                            out.collect(reviewBean);
                        }
                    }
                }
        );
        //reduceDS.print(">>>");
        //TODO 9.关联一下维度表 课程维度表
        SingleOutputStreamOperator<CourseUserReviewBean> resDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<CourseUserReviewBean>("dim_course_info".toUpperCase()) {
                    @Override
                    public void join(CourseUserReviewBean reviewBean, JSONObject jsonObj) {
                        //System.out.println("Phoenix:"+jsonObj.getString("course_name".toUpperCase()));
                        reviewBean.setCourseName(jsonObj.getString("course_name".toUpperCase()));
                        //System.out.println(reviewBean.getCourseName());
                    }

                    @Override
                    public String getKey(CourseUserReviewBean obj) {
                        return obj.getCourseId();
                    }
                },
                5 * 60, TimeUnit.SECONDS

        );

        //TODO 10.将聚合结果写到ck中
        resDS.print(">>>");
        resDS.addSink(MyClickhouseUtil.getSinkFunction("insert into dws_course_user_review_window values(?,?,?,?,?,?,?,?)"));

        env.execute();

    }
}
